package com.galeno.day09;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import lombok.AllArgsConstructor;
import lombok.Getter;
import lombok.NoArgsConstructor;
import lombok.Setter;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;

/**
 * @author galeno
 * @Title:
 * @Description:
 * @date 2021/10/2610:15
 * FLink ProcessFunction,Fink底层的算子,可以访问flink底层的属性和方法
 * 1.处理数据 element
 * 2.使用状态,只能使用keyedStream
 * 3.使用定时器,只能使用keyedStream
 *
 *现在演示的是没有keyBy的DataStream，只能用了处理数据
 *json解析json对象
 */
public class NonKeyedProcessFunction {
    @Setter@Getter@AllArgsConstructor@NoArgsConstructor
    public static  class User{
        private String name;
        private int age;
        private Double fv;
    }
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStreamSource<String> lines = env.socketTextStream("192.168.77.3", 9999);
        /**
         * 如果使用map,出现异常将异常捕获,然后调用哪个filter方法将有问题的数据源过滤掉
         * 现在使用processfunction在一个方法中实现全部的功能
         */
        SingleOutputStreamOperator<User> res = lines.process(new ProcessFunction<String, User>() {
            @Override
            public void processElement(String value, Context ctx, Collector<User> out) throws Exception {
                try {
                    User user = JSON.parseObject(value, User.class);
                    out.collect(user);
                } catch (Exception e) {
                    e.printStackTrace();
                    //有问题的单独数据记录下来
                }
            }
        });
        res.print();
        env.execute();


    }
}
